package com.tbit.uqbike.webmanager.util.csv;

import cn.hutool.core.io.FileUtil;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.google.common.collect.Queues;
import com.tbit.common.config.OssDefaultConfig;
import com.tbit.uqbike.object.pojo.OssConfig;
import com.tbit.uqbike.webmanager.dao.capital.RideLogDetailsDao;
import com.tbit.uqbike.webmanager.util.aliyun.OssFileUtils;
import com.tbit.uqbike.webmanager.util.csv.config.CsvExportConfig;
import com.tbit.uqbike.webmanager.util.csv.config.ExportThreadPool;
import com.tbit.uqbike.webmanager.util.csv.entity.CsvBaseEntity;
import com.tbit.uqbike.webmanager.util.csv.entity.CsvWrite;
import com.tbit.uqbike.webmanager.util.csv.entity.ExportHandler;
import com.tbit.uqbike.webmanager.util.csv.util.DateUtils;
import com.tbit.uqbike.webmanager.util.csv.util.ZipUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.File;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntSupplier;

@Slf4j
@Component
public class CsvExportUtils {

    @Autowired
    CsvExportConfig config;
    @Autowired
    ExportThreadPool pool;
    @Autowired
    private OssDefaultConfig ossDefaultConfig;
    @Autowired
    private RideLogDetailsDao rideLogDetailsDao;



    /**
     * List导出到csv
     *
     * @param tClass
     * @param parent
     * @param fileName
     * @param list
     * @return
     */
    public <T> File exportCsv(Class<T> tClass, File parent, String fileName, List<T> list) {
        File file = new File(parent, fileName + ".csv");
        CsvWrite csvWrite = new CsvWrite(tClass, file);
        csvWrite.writeHead();
        csvWrite.writeList(list);
        csvWrite.writeSum();
        return file;
    }

    public <T> File exportCsv(Class<T> tClass, File parent, String fileName, List<T> list,Boolean writeSum) {
        File file = new File(parent, fileName + ".csv");
        CsvWrite csvWrite = new CsvWrite(tClass, file);
        csvWrite.writeHead();
        csvWrite.writeList(list);
        if (Boolean.TRUE.equals(writeSum)) {
            csvWrite.writeSum();
        }
        return file;
    }

    public <T> File exportBigCsvUseDynamicDate(ExportHandler<T> handler, Class<T> tClass, IntSupplier countFn,
                                               Date startTime, Date endTime, File parent, String fileName, boolean selectWithBetween) {

        File csvFile = new File(parent, fileName + ".csv");
        CsvWrite csvWrite = new CsvWrite(tClass, csvFile);
        csvWrite.writeHead();
        int selectCount = countFn.getAsInt();
        if (selectCount <= 0) {
            csvWrite.writeSum();
            return csvFile;
        }
//        config.getBigDataQueryCount();
        List<List<Date>> splitList = DateUtils.dynamicSplitQueryTimeByDataVolume(selectCount, config.getBigDataQueryCount(), startTime, endTime, selectWithBetween);
        //数据队列
        BlockingQueue<List<Object>> dataQueue = new ArrayBlockingQueue<>(256);
        //读线程计数器
        CountDownLatch countDownLatch = new CountDownLatch(splitList.size());
        //写线程计数
        CountDownLatch writeCountDownLatch = new CountDownLatch(1);
        //单线程写文件
        //运行标记
        AtomicBoolean runFlag = new AtomicBoolean(true);
        //队列中存在的数据量
        AtomicInteger queueItemNum = new AtomicInteger(0);

        pool.execute(() -> {
            while (runFlag.get()) {
                try {
                    if (dataQueue.isEmpty()) {
                        Thread.sleep(config.getWait());
                    }
                    List<Object> list = new ArrayList<>();
                    Queues.drain(dataQueue, list, dataQueue.size(), 3, TimeUnit.SECONDS);
                    int size = 0;
                    for (Object object : list) {
                        List<Object> objectList = (List<Object>) object;
                        for (Object obj : objectList) {
                            List<Object> resultList = (List<Object>) obj;
                            size += resultList.size();
                            csvWrite.writeList(resultList);
                        }
                    }
                    if (list.size() > 0) {
                        list.clear();
                        list = null;
                        queueItemNum.set(queueItemNum.get() - size);
                    }
//                    Thread.sleep(6000);
                    //如果计数器还有数量，说明还有线程没有结束
                    if (countDownLatch.getCount() <= 0) {
                        if (dataQueue.size() <= 0) {
                            break;
                        }
                    }
                } catch (Exception e) {
                    log.error(Thread.currentThread().getName() + "出现错误,{}", e);
                    runFlag.set(false);
                }
            }
            writeCountDownLatch.countDown();
        });
        //异步多线程读  -XX:+PrintGCDetails
        for (List<Date> split : splitList) {
            pool.execute(() -> {
                try {
                    if (runFlag.get()) {
                        CsvBaseEntity csvBaseEntity = new CsvBaseEntity(split.get(0), split.get(1));
                        //限制队列中的数据量，5万以内
                        while (queueItemNum.get() >= config.getMaxQueueItemNum()){
                            log.info("队列中的数据量queueItemNum:{}",queueItemNum);
                            Thread.sleep(config.getMaxQueueItemNumWait());
                        }
                        //获取要导出的业务数据
                        List data = handler.handle(csvBaseEntity);
                        log.info("数据量：{}，查询时间参数:" + com.tbit.uqbike.webmanager.util.DateUtils.DateToString(split.get(0)) + "/" +
                                com.tbit.uqbike.webmanager.util.DateUtils.DateToString(split.get(1)),data.size());
                        if (!data.isEmpty()) {
                            queueItemNum.set(queueItemNum.get() + data.size());
                            dataQueue.offer(Collections.singletonList(data));
                        }
                    }
                } catch (Exception e) {
                    log.error(Thread.currentThread().getName() + "出现错误,{}", e);
                    runFlag.set(false);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        //阻塞等待写线程结束
        try {
            writeCountDownLatch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        //判断异常
        if (!runFlag.get()) {
            throw new RuntimeException("下载导出异常");
        }
        csvWrite.writeSum();
        return csvFile;

    }

    public <T> File exportBigCsvUseDynamicDate(ExportHandler<T> handler, Class<T> tClass, IntSupplier countFn,
                                               Date startTime, Date endTime, File parent, String fileName) {
        return exportBigCsvUseDynamicDate(handler, tClass, countFn, startTime, endTime, parent, fileName, true);
    }

    /**
     * 多线程时间分割导出
     * 注意：
     * 1、sql有group by不能使用，例如1-10天的数据，查询就一条，分成1-5天，一条数据，6-10天，查询又是一条数据，会导致数据变多
     * 2、多线程导出数据是无序的
     *
     * @param handler
     * @param tClass
     * @param startTime
     * @param endTime
     * @param fileName
     * @param parent
     * @return
     */
    public <T> File exportBigCsv(ExportHandler<T> handler, Class<T> tClass, Date startTime, Date endTime, File parent, String fileName) {
        File file = new File(parent, fileName + ".csv");
        //运行标记
        AtomicBoolean runFlag = new AtomicBoolean(true);
        List<List<Date>> splitList = DateUtils.splitByDay(startTime, endTime, config.getProportion());
        BlockingQueue<List<Object>> dataQueue = new ArrayBlockingQueue<>(1024);//数据队列
        CountDownLatch countDownLatch = new CountDownLatch(splitList.size());//读线程计数器
        CountDownLatch writeCountDownLatch = new CountDownLatch(1);//写线程计数
        CsvWrite csvWrite = new CsvWrite(tClass, file);
        csvWrite.writeHead();

        //单线程写文件
        pool.execute(() -> {

            while (runFlag.get()) {
                try {
                    if (dataQueue.isEmpty()) {
                        Thread.sleep(config.getWait());
                    }


                    List<Object> list = new ArrayList<>();
                    Queues.drain(dataQueue, list, dataQueue.size(), 3, TimeUnit.SECONDS);
                    for (Object object : list) {
                        List<Object> objectList = (List<Object>) object;
                        for (Object obj : objectList) {
                            List<Object> resultList = (List<Object>) obj;
                            csvWrite.writeList(resultList);
                        }
                    }

                    //如果计数器还有数量，说明还有线程没有结束
                    if (countDownLatch.getCount() <= 0) {
                        if (dataQueue.size() <= 0) {
                            break;
                        }
                    }
                } catch (Exception e) {
                    log.error(Thread.currentThread().getName() + "出现错误,{}", e);
                    runFlag.set(false);
                }
            }
            writeCountDownLatch.countDown();
        });

        //异步多线程读
        for (List<Date> split : splitList) {
            pool.execute(() -> {
                try {
                    if (runFlag.get()) {
                        CsvBaseEntity csvBaseEntity = new CsvBaseEntity(split.get(0), split.get(1));
                        //获取要导出的业务数据
                        List data = handler.handle(csvBaseEntity);
                        if (!data.isEmpty()) {
                            dataQueue.offer(Collections.singletonList(data));
                        }
                    }
                } catch (Exception e) {
                    log.error(Thread.currentThread().getName() + "出现错误,{}", e);
                    runFlag.set(false);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        //阻塞等待写线程结束
        try {
            writeCountDownLatch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }

        //判断异常
        if (!runFlag.get()) {
            throw new RuntimeException("下载导出异常");
        }
        csvWrite.writeSum();
        return file;
    }


    /**
     * 多线程PageHelper分页导出
     * 注意：
     * 1、对于小的偏移量，直接用 limit 查询没有什么问题。随着数据量的增大，越往后分页，limit 语句的偏移量越大，速度也会明显变慢。
     * 2、多线程导出数据是无序的
     *
     * @param handler   处理数据的接口
     * @param tClass    导出的类型
     * @param startTime 开始
     * @param endTime   结束
     * @param fileName  文件名
     * @return
     */
    public <T> File export(ExportHandler<T> handler, Class<T> tClass, Date startTime, Date endTime, File parent, String fileName) {
        File file = new File(parent, fileName + ".csv");

        int pageSize = 10000;
        CsvBaseEntity base = new CsvBaseEntity(startTime, endTime);
        PageHelper.startPage(1, pageSize);
        List<T> firstList = handler.handle(base);
        PageInfo<T> pageInfo = new PageInfo(firstList);

        CsvWrite csvWrite = new CsvWrite(tClass, file);
        csvWrite.writeHead();
        if (!firstList.isEmpty()) {
            csvWrite.writeList(firstList);
        }

        if (pageInfo.getPages() > 1) {
            //运行标记
            AtomicBoolean runFlag = new AtomicBoolean(true);
            BlockingQueue<List<Object>> dataQueue = new ArrayBlockingQueue<>(1024);//数据队列
            CountDownLatch countDownLatch = new CountDownLatch(pageInfo.getPages() - 1);//读线程计数器
            CountDownLatch writeCountDownLatch = new CountDownLatch(1);//写线程计数
            //单线程写文件
            pool.execute(() -> {
                while (runFlag.get()) {
                    try {
                        if (dataQueue.isEmpty()) {
                            Thread.sleep(config.getWait());
                        }

                        List<Object> list = new ArrayList<>();
                        Queues.drain(dataQueue, list, dataQueue.size(), 3, TimeUnit.SECONDS);
                        for (Object object : list) {
                            List<Object> objectList = (List<Object>) object;
                            for (Object obj : objectList) {
                                List<Object> resultList = (List<Object>) obj;
                                csvWrite.writeList(resultList);
                            }
                        }

                        //如果计数器还有数量，说明还有线程没有结束
                        if (countDownLatch.getCount() <= 0) {
                            if (dataQueue.size() <= 0) {
                                break;
                            }
                        }
                    } catch (Exception e) {
                        log.error(Thread.currentThread().getName() + "出现错误,{}", e);
                        runFlag.set(false);
                    }
                }
                writeCountDownLatch.countDown();
            });

            for (int i = 2; i < pageInfo.getPages() + 1; i++) {
                int finalI = i;
                pool.execute(() -> {
                    try {
                        if (runFlag.get()) {
                            //获取要导出的业务数据
                            PageHelper.startPage(finalI, pageSize, false);
                            List data = handler.handle(base);
                            if (!data.isEmpty()) {
                                dataQueue.offer(Collections.singletonList(data));
                            }
                        }
                    } catch (Exception e) {
                        log.error(Thread.currentThread().getName() + "出现错误,{}", e);
                        runFlag.set(false);
                    } finally {
                        countDownLatch.countDown();
                    }
                });
            }

            //阻塞等待写线程结束
            try {
                writeCountDownLatch.await();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }

            //判断异常
            if (!runFlag.get()) {
                throw new RuntimeException("导出异常");
            }
        }
        csvWrite.writeSum();
        return file;
    }


    /**
     * 获取临时的目录
     *
     * @return
     */
    public File getTempFile() {
        File file = new File(config.getTempPath(), UUID.randomUUID().toString().replaceAll("-", ""));
        if (!FileUtil.exist(file)) {
            FileUtil.mkdir(file);
        }
        return file;
    }

    /**
     * 获取临时的目录
     *
     * @return
     */
    public File getTempFileByName(String fileName) {
        File file = new File(config.getTempPath(), fileName);
        if (!FileUtil.exist(file)) {
            FileUtil.mkdir(file);
        }
        return file;
    }


    /**
     * 压缩文件并上传到oss
     *
     * @param parent
     * @param fileName
     * @param files
     * @return
     */
    public String uploadZip(File parent, String fileName, File... files) {
        fileName = fileName + ".zip";
        File zip = zip(parent, fileName, files);
        //上传oss
        OssConfig ossConfig = new OssConfig(ossDefaultConfig.getEndpoint(),
                ossDefaultConfig.getAccessKeyId(),
                ossDefaultConfig.getAccessKeySecret(), config.getBucketName());
        return OssFileUtils.uploadZip(ossConfig, fileName, zip);
    }


    /**
     * 文件压缩
     *
     * @param parent
     * @param fileName
     * @param files
     * @return
     */
    private File zip(File parent, String fileName, File... files) {
        List<File> srcFile = new ArrayList();
        for (File file : files) {
            srcFile.add(file);
        }
        File zipfile = new File(parent, fileName);
        ZipUtil.zipFiles(srcFile, zipfile);
        return zipfile;
    }



    public <T> File exportBigCsvUseDynamicId(ExportHandler<T> handler, Class<T> tClass, int count,
                                             File parent, String fileName,
                                             Integer maxId,
                                             Integer minId) {

        File csvFile = new File(parent, fileName + ".csv");
        CsvWrite csvWrite = new CsvWrite(tClass, csvFile);
        csvWrite.writeHead();
//        config.getBigDataQueryCount();
        List<Map<String, Integer>> idMapList = this.dynamicSplitQueryById(count, maxId, minId,config.getBigDataQueryCount());
        //数据队列
        BlockingQueue<List<Object>> dataQueue = new ArrayBlockingQueue<>(256);
        //读线程计数器
        CountDownLatch countDownLatch = new CountDownLatch(idMapList.size());
        //写线程计数
        CountDownLatch writeCountDownLatch = new CountDownLatch(1);
        //单线程写文件
        //运行标记
        AtomicBoolean runFlag = new AtomicBoolean(true);
        //队列中存在的数据量
        AtomicInteger queueItemNum = new AtomicInteger(0);

        pool.execute(() -> {
            while (runFlag.get()) {
                try {
                    if (dataQueue.isEmpty()) {
                        Thread.sleep(config.getWait());
                    }
                    List<Object> list = new ArrayList<>();
                    Queues.drain(dataQueue, list, dataQueue.size(), 3, TimeUnit.SECONDS);
                    int size = 0;
                    for (Object object : list) {
                        List<Object> objectList = (List<Object>) object;
                        for (Object obj : objectList) {
                            List<Object> resultList = (List<Object>) obj;
                            size += resultList.size();
                            csvWrite.writeList(resultList);
                        }
                    }
                    if (list.size() > 0) {
                        list.clear();
                        list = null;
                        queueItemNum.set(queueItemNum.get() - size);
                    }
                    //如果计数器还有数量，说明还有线程没有结束
                    if (countDownLatch.getCount() <= 0) {
                        if (dataQueue.size() <= 0) {
                            break;
                        }
                    }
                } catch (Exception e) {
                    log.error(Thread.currentThread().getName() + "出现错误,{}", e);
                    runFlag.set(false);
                }
            }
            writeCountDownLatch.countDown();
        });
        //异步多线程读  -XX:+PrintGCDetails
        for (Map<String, Integer> idMap : idMapList) {
            pool.execute(() -> {
                try {
                    if (runFlag.get()) {
                        CsvBaseEntity csvBaseEntity = new CsvBaseEntity(idMap.get("startId"), idMap.get("endId"));
                        //限制队列中的数据量，5万以内
                        while (queueItemNum.get() >= config.getMaxQueueItemNum()){
                            log.info("队列中的数据量queueItemNum:{}",queueItemNum);
                            Thread.sleep(config.getMaxQueueItemNumWait());
                        }
                        //获取要导出的业务数据
                        List data = handler.handle(csvBaseEntity);
                        log.info("数据量：{}，查询id区间参数:" + idMap.get("startId") + "/" +
                                idMap.get("endId"),data.size());
                        if (!data.isEmpty()) {
                            queueItemNum.set(queueItemNum.get() + data.size());
                            dataQueue.offer(Collections.singletonList(data));
                        }
                    }
                } catch (Exception e) {
                    log.error(Thread.currentThread().getName() + "出现错误,{}", e);
                    runFlag.set(false);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        //阻塞等待写线程结束
        try {
            writeCountDownLatch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        //判断异常
        if (!runFlag.get()) {
            throw new RuntimeException("下载导出异常");
        }
        return csvFile;

    }

    private List<Map<String,Integer>> dynamicSplitQueryById(int count,Integer maxId,Integer minId,int everySize){
        List<Map<String,Integer>> idMapList = new ArrayList<>();
        //分批数
        int listNum = (int) Math.ceil((double) count / everySize);
        //最大最小id的差值
        int idDiffer = maxId - minId;
        //每一批的id数量
        int size = (int) Math.ceil((double) idDiffer / listNum);
        for (int j = 1; j <=listNum; j++) {
            Map<String,Integer> idMap = new HashMap<>();
            idMap.put("startId",minId);
            idMap.put("endId",minId + size);
            idMapList.add(idMap);
            minId = minId + size;
        }
        return idMapList;
    }

}
